//*********************************************************
//
//    Copyright (c) Microsoft. All rights reserved.
//    This code is licensed under the Apache License, Version 2.0.
//    THIS CODE IS PROVIDED *AS IS* WITHOUT WARRANTY OF
//    ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING ANY
//    IMPLIED WARRANTIES OF FITNESS FOR A PARTICULAR
//    PURPOSE, MERCHANTABILITY, OR NON-INFRINGEMENT.
//
//*********************************************************

namespace Microsoft.Research.ProvenanceSubscriberMessageParser
{
    using System;
    using System.Collections.Generic;
    using System.Collections.ObjectModel;
    using System.Threading;
    using Microsoft.Research.ProvenanceSubscriberCommon;
    using Microsoft.Research.ProvenanceSubscriberCommon.Interfaces;
    using Microsoft.Research.ProvenanceSubscriberCommon.Models;
    using Microsoft.Research.ScientificWorkflow.TridentUtilities;
    using Microsoft.Research.eResearch.Blackboard.Common;
    using Microsoft.Research.eResearch.Common.Eventing;
    using Microsoft.Research.eResearch.Subscriber;
    using Microsoft.Research.eResearch.Subscriber.Interfaces;
    using System.Xml;
    using System.IO;
    using System.Globalization;

    /// <summary>
    /// This file contains logic to integrate with blackboard host 
    /// and then shield Blackboard messages.
    /// </summary>
    public class BlackboardMessageParser : IProvenanceService
    {
        /// <summary>
        /// Field indicating whether the monitor service has been started.
        /// </summary>
        private bool started;

        /// <summary>
        /// Instance of the blackboard subscriber
        /// </summary>
        private BlackboardSubscriber blackboardSubscriber;

        /// <summary>
        /// This will hold all the message parsers.
        /// </summary>
        private static Collection<AbstractMessageParser> messageParsers = new Collection<AbstractMessageParser>();

        /// <summary>
        /// Thread used to store Provenance Info into Provenance Db
        /// </summary>
        private Thread provenanceThread;

        /// <summary>
        /// Holds all Incomming message from BlackBoard
        /// </summary>
        private static Queue<BlackboardMessage> ProvenanceQueue;

        /// <summary>
        /// Flag holds true when performance Thread is Running
        /// </summary>
        private static bool provenanceThreadRunning;

        /// <summary>
        /// Locks the Parsing Section
        /// </summary>
        private static object provenanceLock = new object();

        #region Constructors.

        /// <summary>
        /// Constructor.
        /// </summary>
        public BlackboardMessageParser()
        {
            this.blackboardSubscriber = new BlackboardSubscriber();
            this.blackboardSubscriber.BlackboardMessageReceived += this.OnBlackboardMessageReceived;
        }

        #endregion

        #region Events.

        public event EventHandler<ProvenanceModelEventArgs> ProvenanceMessage;

        public event EventHandler<ProvenanceModelEventArgs> InputOutputMessage;

        public event EventHandler<ProvenanceModelEventArgs> PerformanceMessage;

        public event EventHandler<ProvenanceModelEventArgs> ProcessingStatusMessage;

        public event EventHandler<ProvenanceModelEventArgs> DataProductMessage;

        #endregion

        /// <summary>
        /// Raise Provenance Event.
        /// </summary>
        /// <param name="provenanceModel"></param>
        internal void RaiseProvenanceEvent(ProvenanceInformationModel provenanceModel)
        {
            ProvenanceModelEventArgs provEventAgrs = new ProvenanceModelEventArgs(provenanceModel);

            this.ProvenanceMessage.Fire(this, provEventAgrs);
        }

        /// <summary>
        /// Raise InputOutPut Event.
        /// </summary>
        /// <param name="provenanceModel"></param>
        internal void RaiseInputOutEvent(ProvenanceInformationModel provenanceModel)
        {
            ProvenanceModelEventArgs provEventAgrs = new ProvenanceModelEventArgs(provenanceModel);

            this.InputOutputMessage.Fire(this, provEventAgrs);
        }

        /// <summary>
        /// Raise Performance Event.
        /// </summary>
        /// <param name="provenanceModel"></param>
        internal void RaisePerformanceEvent(ProvenanceInformationModel provenanceModel)
        {
            ProvenanceModelEventArgs provEventAgrs = new ProvenanceModelEventArgs(provenanceModel);

            this.PerformanceMessage.Fire(this, provEventAgrs);
        }

        /// <summary>
        /// Raise ProcessingStatus Event.
        /// </summary>
        /// <param name="provenanceModel"></param>
        internal void RaiseProcessingStatusEvent(ProvenanceInformationModel provenanceModel)
        {
            ProvenanceModelEventArgs provEventAgrs = new ProvenanceModelEventArgs(provenanceModel);

            this.ProcessingStatusMessage.Fire(this, provEventAgrs);
        }

        /// <summary>
        /// Raise ProcessingStatus Event.
        /// </summary>
        /// <param name="provenanceModel"></param>
        internal void RaiseDataProductEvent(ProvenanceInformationModel provenanceModel)
        {
            ProvenanceModelEventArgs provEventAgrs = new ProvenanceModelEventArgs(provenanceModel);

            this.DataProductMessage.Fire(this, provEventAgrs);
        }

        /// <summary>
        /// This function is used to register all the parsers present.
        /// </summary>
        private void RegisterParsers()
        {
            // Add Processing status message Parser.
            messageParsers.Add(new ProcessingStatusMessageParser(this));
            messageParsers.Add(new PerformanceDetailParser(this));
            messageParsers.Add(new InputOutputParameterParser(this));
            messageParsers.Add(new DataProductMessageParser(this));
        }

        /// <summary>
        /// Deques the Message and store it in the Provenance Db
        /// </summary>
        private static void PopProvenanceInfo()
        {
            while (provenanceThreadRunning)
            {

                BlackboardMessage message = null;

                try
                {
                    lock (BlackboardMessageParser.provenanceLock)
                    {
                        if (ProvenanceQueue.Count > 0)
                        {
                            message = ProvenanceQueue.Peek();
                            if (message != null)
                            {
                                foreach (AbstractMessageParser parser in messageParsers)
                                {
                                    parser.Parse(message);
                                }
                            }

                            ProvenanceQueue.Dequeue();
                        }
                        else
                        {
                            Thread.Sleep(500);
                        }
                    }
                }
                catch (Exception ex)
                {
                    string messageXml = MessageToXML(message);
                    TridentCustomException customException = new TridentCustomException(string.Format(CultureInfo.InvariantCulture, ProvenanceResourceManager.GetString("ErrorMessageDequeue"), messageXml), ex);
                    TridentErrorHandler.HandleUIException(customException);
                }
            }
        }

        #region Public Methods.

        /// <summary>
        /// Start the subscriber if not already started.
        /// </summary>
        public void Start()
        {
            try
            {
                if (!this.started)
                {
                    // Register parsers.
                    this.RegisterParsers();

                    this.blackboardSubscriber.Start();

                    SubscriptionProfile profile = new SubscriptionProfile();
                    profile.Add(new SubscriptionItem("trident:InstanceId"));
                    profile.Add(new SubscriptionItem("trident:WorkflowType"));
                    profile.Add(new SubscriptionItem("trident:ActivityState"));
                    profile.Add(new SubscriptionItem("trident:ActivityName"));
                    profile.Add(new SubscriptionItem("trident:ActivityLabel"));
                    profile.Add(new SubscriptionItem("trident:WorkflowState"));
                    profile.Add(new SubscriptionItem("trident:EventOrder"));
                    profile.Add(new SubscriptionItem("trident:CLRType"));
                    profile.Add(new SubscriptionItem("trident:ActivityPropertyValue"));
                    profile.Add(new SubscriptionItem("trident:FieldName"));
                    profile.Add(new SubscriptionItem("trident:ActivityPropertyDirection"));
                    profile.Add(new SubscriptionItem("trident:CpuResource"));
                    profile.Add(new SubscriptionItem("trident:MemoryResource"));
                    profile.Add(new SubscriptionItem("trident:EventTimestamp"));
                    profile.Add(new SubscriptionItem("time:DateTimeDescription"));
                    profile.Add(new SubscriptionItem("trident:JobId"));
                    profile.Add(new SubscriptionItem("trident:WorkflowName"));
                    profile.Add(new SubscriptionItem("trident:ExceptionDetails"));
                    profile.Add(new SubscriptionItem("trident:UserData"));
                    profile.Add(new SubscriptionItem("trident:UserDataKey"));
                    profile.Add(new SubscriptionItem("trident:ActivityPropertyType"));
                    profile.Add(new SubscriptionItem("trident:IterationValue"));
                    profile.Add(new SubscriptionItem("trident:DataProductID"));
                    profile.Add(new SubscriptionItem("trident:OutputDataProductID"));
                    this.blackboardSubscriber.Subscribe(profile);

                    provenanceThreadRunning = true;

                    this.provenanceThread = new Thread(new ThreadStart(BlackboardMessageParser.PopProvenanceInfo));
                    ProvenanceQueue = new Queue<BlackboardMessage>();
                    this.provenanceThread.Start();
                    started = true;
                }
            }
            catch
            {
                throw;
            }
        }

        /// <summary>
        /// Stop the subscriber if already started.
        /// </summary>
        public void StopSubscriber()
        {
            try
            {
                if (this.started)
                {
                    if (ProvenanceQueue.Count > 0)
                    {
                        throw new TridentCustomException(TridentErrorConstants.ErrorNumber1100000016, ProvenanceResourceManager.GetString("TrayAppCloseWarning"));
                    }
                    else
                    {
                        this.blackboardSubscriber.Stop();
                        this.started = false;
                        this.provenanceThread.Abort();
                        ProvenanceQueue = null;
                        provenanceThreadRunning = false;
                    }
                }
            }
            catch (TridentCustomException exp)
            {
                TridentErrorHandler.ThrowTridentCustomException(TridentErrorConstants.ErrorNumber1100000016, exp.CustomMessage.ToString());
            }

            catch (Exception exp)
            {
                TridentErrorHandler.HandleUIException(exp);
            }
        }

        #endregion

        #region Event Handlers

        /// <summary>
        /// Event handler for black board message recieved event.
        /// </summary>
        /// <param name="sender">Instance of the subscriber.</param>
        /// <param name="e">Instance containing the blackboard message.</param>
        private void OnBlackboardMessageReceived(object sender, BlackboardMessageReceivedEventArgs e)
        {
            try
            {
                // Add the Incomming message into the Queue
                ProvenanceQueue.Enqueue(e.IncomingMessage);
            }
            catch (Exception ex)
            {
                string messageXml = MessageToXML(e.IncomingMessage);
                TridentCustomException customException = new TridentCustomException(string.Format(CultureInfo.InvariantCulture, ProvenanceResourceManager.GetString("ErrorMessageEnQueue"), messageXml), ex);
                TridentErrorHandler.HandleUIException(customException);
            }
        }

        /// <summary>
        /// Converts the BlackBoard Message to String
        /// </summary>
        /// <param name="blackBoardMessage">BlackboardMessage</param>
        /// <returns>string</returns>
        private static string MessageToXML(BlackboardMessage blackBoardMessage)
        {
            string messageContent = string.Empty;
            try
            {
                if (blackBoardMessage == null)
                {
                    return messageContent;
                }
                StringWriter stringWriter = new StringWriter(CultureInfo.InvariantCulture);
                XmlTextWriter myXmlWriter = new XmlTextWriter(stringWriter);
                blackBoardMessage.WriteXml(myXmlWriter);
                messageContent = stringWriter.ToString();
                myXmlWriter.Close();
                return messageContent;
            }

            catch (XmlException ex)
            {
                TridentErrorHandler.HandleUIException(ex);
            }
            return messageContent;
        }

        #endregion
    }
}
